package flow;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

public class FlowDemo {

    public static void main(String[] args) throws InterruptedException {

        // 1. 定义一个发布者，发布数据
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();


        // 2. 定义一个订阅者，订阅数据发布
        Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() {

            private Flow.Subscription subscription;

            // 订阅时执行这个回调
            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                System.out.println(Thread.currentThread() + "订阅开始了：" + subscription);
                this.subscription = subscription;
                // 订阅时，请求一条数据
                subscription.request(1);
            }

            // 下一个元素到达时执行这个回调
            @Override
            public void onNext(String item) {
                System.out.println(Thread.currentThread() + "订阅者接收到数据：" + item);
                // 接收到一条数据后，请求下一条数据
                subscription.request(1);
            }

            // 错误发生时执行这个回调
            @Override
            public void onError(Throwable throwable) {
                System.out.println(Thread.currentThread() + "订阅者接收到错误信号：" + throwable);
            }

            // 完成时执行这个回调
            @Override
            public void onComplete() {
                System.out.println(Thread.currentThread() + "订阅者接收到完成信号");
            }
        };

        // 绑定发布者和订阅者
        publisher.subscribe(subscriber);

        for (int i = 0; i < 10; i++) {
            // 发布数据，publish发布的所有数据都缓存在它的buffer区
            System.out.println(Thread.currentThread() + "发布者发布数据");
            publisher.submit("p-" + i);
        }

        // 数据发布完成后关闭，触发订阅者的onComplete事件
        publisher.close();

        Thread.sleep(2000);
    }
}
